package cn.soulic.tools.redis.migration;

import cn.soulic.tools.redis.migration.config.RedisMigrationProperties;
import cn.soulic.tools.redis.migration.handler.RedisMigrationHandler;
import io.lettuce.core.KeyScanCursor;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.ScanArgs;
import io.lettuce.core.ScanCursor;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.sync.RedisCommands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * 执行器
 *
 * @author Soulic Yang
 * @since 1.0.0
 */
@Component
public class RedisMigrationRunner implements CommandLineRunner, DisposableBean {
    private final Logger log = LoggerFactory.getLogger(RedisMigrationRunner.class);

    @Resource
    private RedisMigrationProperties redisMigrationProperties;

    @Resource
    private Map<String, RedisMigrationHandler> handlers;

    @Resource
    private RedisClient sourceRedisClient;

    @Resource
    private RedisClient targetRedisClient;

    @Resource
    private StatefulRedisConnection<String, String> sourceConnection;

    @Resource
    private StatefulRedisConnection<String, String> targetConnection;

    @Resource
    private RedisCommands<String, String> sourceCommands;

    @Resource
    private RedisCommands<String, String> targetCommands;

    @Override
    public void run(String... args) throws Exception {
        RedisAsyncCommands<String, String> async = sourceConnection.async();
        RedisFuture<String> info = async.info("Keyspace");
        try {
            String s = info.get();
            String[] split = s.split("\r\n");
            int keyCount = 0;
            String dbPrefix = String.format("db%s:", redisMigrationProperties.getSource().getDatabase());
            for (String str : split) {
                if (str.startsWith(dbPrefix)) {
                    String[] dbInfo = str.replace(dbPrefix, "").split(",");
                    for (String property : dbInfo) {
                        if (property.startsWith("keys=")) {
                            String value = property.replace("keys=", "");
                            keyCount = Integer.parseInt(value);
                            break;
                        }
                    }
                    break;
                }
            }

            if (keyCount == 0) {
                return;
            }

            doMigrate(keyCount);
        } catch (InterruptedException | ExecutionException e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }

    private void doMigrate(int keyCount) {
        ScanArgs matchesAll = ScanArgs.Builder.matches("*").limit(redisMigrationProperties.getMaxCount());
        ScanCursor cursor = ScanCursor.INITIAL;

        do {
            KeyScanCursor<String> scanResult = sourceCommands.scan(cursor, matchesAll);
            List<String> keys = scanResult.getKeys();
            handleKeys(keys);
            cursor = scanResult;
        } while (!cursor.isFinished());
    }

    private void handleKeys(List<String> keys) {
        for (String key : keys) {
            String type = sourceCommands.type(key);
            log.info("key: {}", key);
            RedisMigrationHandler handler = handlers.get(type);
            if (handler != null) {
                handler.handle(key);
            } else {
                log.info("key[{}] type[{}]", key, type);
            }
        }
    }

    public void destroy() {
        targetCommands.shutdown(true);
        targetConnection.close();
        targetRedisClient.shutdown();

        sourceCommands.shutdown(true);
        sourceConnection.close();
        sourceRedisClient.shutdown();
    }
}
